mod prelude;

use std::{collections::BTreeMap, path::PathBuf, sync::Arc};

use ::oracle::sql_type::OracleType;
use ahash::AHashMap;
use chrono::{FixedOffset, Local, SubsecRound};
use lazy_static::lazy_static;
use parking_lot::Condvar;
use prelude::{OracleManager, QueryTable};
use tdc::{Action, DatabaseConfig, GlobalConfig, Rule, TableInfo};
use tokio::{
    fs::OpenOptions,
    io::{AsyncWriteExt, BufWriter},
    runtime,
    sync::Mutex,
};
// use tokio::{fs::File, io::{BufWriter, AsyncWriteExt}};

const ROWID: &str = "ROWID";
const ROWID_ALIAS: &str = "$ROWID";

macro_rules! paging {
    ($($sql:expr),*) => {
        format!("SELECT a.* FROM (SELECT ROWID AS \"$ROWID\", t.*, ROWNUM AS rn FROM ( {} ) t WHERE ROWNUM <= :1 ) a WHERE rn >= :2", format_args!($($sql),*))
    };
}

// lazy_static! {
//     static ref FILE_CACHE: Mutex<HashMap<u32, File>> = Mutex::new(HashMap::new());
// }
lazy_static! {
    pub static ref SHARE_GLOBAL_AREA_MUTEX_PAIR: Arc<(
        parking_lot::lock_api::Mutex<parking_lot::RawMutex, AHashMap<u32, BTreeMap<String, u32>>>,
        Condvar
    )> = Arc::new((parking_lot::Mutex::new(AHashMap::new()), Condvar::new()));
}

// pub struct CacheTable {
//     pub count: usize,
//     pub counter: usize,
//     pub page_count: usize,
//     pub owner: String,
//     pub table_name: String,
// }

// impl CacheTable {
//     pub fn new() -> Self {
//         CacheTable {
//             count: 0,
//             counter: 0,
//             page_count: 0,
//             owner: String::new(),
//             table_name: String::new(),
//         }
//     }
// }

// 转换为UTC时区
#[inline]
fn to_utc(tz: Option<chrono::DateTime<FixedOffset>>) -> String {
    // 强制转换为UTC时区
    match tz {
        // 是否去掉精度
        Some(ts) => ts.naive_utc().trunc_subsecs(0).to_string(),
        None => String::new(),
    }
}

// 截断毫秒
#[inline]
fn trunc_subsecs(ts: Option<oracle::sql_type::Timestamp>) -> String {
    if let Some(ts0) = ts {
        match ts0.and_prec(0) {
            Ok(ts1) => return ts1.to_string(),
            Err(e) => {
                log::error!("oracle::timestamp::trunc_subsecs error: {e}");
                return ts0.to_string();
            }
        }
    }
    String::new()
}

// fn to_crc64(chunk: &[u8]) -> u64 {
//     let mut c = crc64fast::Digest::new();
//     c.write(chunk);
//     c.sum64()
// }

// 计算每一行的crc码
fn checksum(row: oracle::Row, primary_keys: Vec<String>) -> (u32, String, u32) {
    let mut input: Vec<Vec<u8>> = Vec::new();
    let mut primary_key_map = AHashMap::new();
    // let mut crc_map = AHashMap::new();
    for col in row.column_info() {
        let s_col = col.name();
        let x_col = s_col.to_owned();
        match col.oracle_type().to_owned() {
            OracleType::Rowid => {
                let value: String = row.get(s_col).unwrap();
                if s_col == ROWID || s_col == ROWID_ALIAS {
                    if primary_keys.contains(&x_col) || primary_keys.len() == 0 {
                        primary_key_map.insert(ROWID.to_owned(), value);
                    }
                } else {
                    input.push(value.into_bytes());
                }
            }
            // 不做任何处理
            OracleType::Raw(_)
            | OracleType::IntervalDS(_, _)
            | OracleType::IntervalYM(_)
            | OracleType::Date
            | OracleType::Varchar2(_)
            | OracleType::NVarchar2(_)
            | OracleType::Number(_, _)
            | OracleType::Float(_)
            | OracleType::BinaryFloat
            | OracleType::BinaryDouble => {
                let value: Option<String> = row.get(s_col).unwrap();
                // log::debug!("{}: {:?}", col.name(), value);
                let value = value.unwrap_or_default();
                if primary_keys.contains(&x_col) {
                    primary_key_map.insert(x_col, value.clone());
                }
                input.push(value.into_bytes());
            }

            OracleType::Long | OracleType::CLOB | OracleType::NCLOB | OracleType::Xml => {
                // 字符(大字段)
                let value: Option<String> = row.get(s_col).unwrap();
                let value = value.unwrap_or_default();
                if primary_keys.contains(&x_col) {
                    primary_key_map.insert(x_col, value.clone());
                }
                input.push(value.into_bytes());
            }

            OracleType::LongRaw | OracleType::BLOB | OracleType::BFILE => {
                // 二进制(大字段)
                let value: Option<Vec<u8>> = row.get(s_col).unwrap();
                // log::debug!("{}: {:?}", s_col, value);
                input.push(value.unwrap_or_default());
            }

            OracleType::Char(_) | OracleType::NChar(_) => {
                // 去掉后面的空格
                let value: Option<String> = row.get(s_col).unwrap();
                let value = value.unwrap_or_default().trim_end().to_owned();
                if primary_keys.contains(&x_col) {
                    primary_key_map.insert(x_col, value.clone());
                }
                input.push(value.into_bytes());
            }
            OracleType::Timestamp(_) => {
                // 去掉精度
                let value = match row.get(col.name()) {
                    Ok(value) => trunc_subsecs(value),
                    Err(e) => {
                        log::error!("{e}");
                        String::new()
                    }
                };
                if primary_keys.contains(&x_col) {
                    primary_key_map.insert(x_col, value.clone());
                }
                input.push(value.into_bytes());
            }

            OracleType::TimestampTZ(_) | OracleType::TimestampLTZ(_) => {
                // 强制转换为UTC时区
                // 已通过+0转换为date类型
                let value = match row.get(col.name()) {
                    Ok(value) => to_utc(value),
                    Err(e) => {
                        log::error!("{e}");
                        String::new()
                    }
                };
                if primary_keys.contains(&x_col) {
                    primary_key_map.insert(x_col, value.clone());
                }
                input.push(value.into_bytes());
            }
            _ => {}
        }
    }

    let unique_values = if primary_keys.len() == 0 {
        // 是否为无主键表
        vec![primary_key_map.get(ROWID).unwrap().as_str()]
    } else {
        primary_keys
            .iter()
            .map(|k| primary_key_map.get(k.as_str()).unwrap().as_str())
            .collect::<Vec<&str>>()
    };
    // log::debug!("{:?} => {:?}", unique_values, input);

    // 数据checksum
    let checksum = match bincode::encode_to_vec(&input, bincode::config::standard()) {
        Ok(encoded) => crc32fast::hash(&encoded),
        Err(e) => {
            log::error!("Encode Error: {e}");
            0
        }
    };

    let key_crc = match bincode::encode_to_vec(&unique_values, bincode::config::standard()) {
        Ok(encoded) => crc32fast::hash(&encoded),
        // 计算并取模
        // return (crc32fast::hash(&encoded), format!("{},{:08x}", unique_values.join("-"), checksum));
        Err(e) => {
            log::error!("Encode Error: {e}");
            0
        }
    };
    let key = unique_values.join("-");
    (key_crc, key.clone(), checksum)
}

// 查询数据并计算crc码
#[inline]
async fn query_and_checksum(
    page: usize,
    table: TableInfo,
    global_config: GlobalConfig,
    manager: &OracleManager,
) -> usize {
    let primary_keys = table.keys.clone();
    let scols = if global_config.with_long_type {
        table.query_col_names()
    } else {
        table.query_nolong_col_names()
    }
    .join(",");

    // 表无主键且无非空唯一索引
    // 最后使用rowid
    let paging_sql = paging!(
        "select /*+ RESULT_CACHE */ {} from {:?}.{:?} order by \"{}\"",
        scols,
        &table.schema_name,
        &table.table_name,
        if primary_keys.len() == 0 {
            "ROWID".to_owned()
        } else {
            primary_keys.join("\", \"")
        }
    );

    let start = (page - 1) * global_config.page_size + 1;
    let end = page * global_config.page_size;

    log::debug!("SQL> {paging_sql} << ~~ [{}, {}]", end, start);
    let now = Local::now();
    let mut fetched_count = 0;
    match manager.get_conn().query(&paging_sql, &[&end, &start]) {
        Ok(rows) => {
            for row_result in rows {
                match row_result {
                    Ok(row) => {
                        fetched_count += 1;
                        // key_crc: row_crc
                        let (key_crc, key, value) = checksum(row, primary_keys.clone());
                        let seq = key_crc % global_config.buckets as u32;

                        {
                            let &(ref lock, ref cvar) = &*SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                            let mut map = lock.lock();
                            // Acquired Lock: ReadWrite
                            map.entry(seq).or_insert(BTreeMap::new()).insert(key, value);
                            // 通知其他等待的线程
                            cvar.notify_all();
                            // Released Lock
                        }
                    }
                    Err(e) => log::error!("convert failed, {e}"),
                }
            }
        }
        Err(e) => log::error!("ResultSet Error: {e}"),
    }

    let now = Local::now() - now;
    log::info!(
        "Table [{}.{}] rows fetched: {} ({:.3} sec)",
        table.schema_name,
        table.table_name,
        fetched_count,
        (now.num_milliseconds() as f64 / 1000.0) as f64
    );

    fetched_count
}

// #[inline]
// async fn write_crc_file(
//     global_config: GlobalConfig,
//     mut table: TableInfo,
//     crc_map: AHashMap<u32, BTreeMap<String, String>>,
// ) -> AHashMap<String, usize> {
//     for (key, value) in crc_map {
//         let file_name = global_config
//             .tmpdir
//             .join(format!("{}.{}_{}.csv", table.owner, table.table_name, key));
//         table
//             .crc_file_names
//             .insert(file_name.clone().to_str().unwrap().to_string(), 1);

//         match OpenOptions::new()
//             .append(true)
//             .create(true)
//             .open(file_name)
//             .await
//         {
//             Ok(file) => {
//                 let mut writer = BufWriter::new(file);
//                 let mut contents = Vec::new();
//                 for (k, v) in value {
//                     contents.push(format!("{k},{v}"));
//                 }

//                 if let Err(e) = writer
//                     .write_all(format!("{}\n", contents.join("\n")).as_bytes())
//                     .await
//                 {
//                     log::error!("{e}");
//                 }
//                 if let Err(e) = writer.flush().await {
//                     log::error!("{e}");
//                 }
//             }
//             Err(e) => log::error!("{e}"),
//         }
//     }

//     table.crc_file_names
// }

#[inline]
async fn write_contents(path: PathBuf, contents: &[u8]) {
    match OpenOptions::new()
        .append(true)
        .create(true)
        .open(path)
        .await
    {
        Ok(file) => {
            let mut writer = BufWriter::new(file);

            if let Err(e) = writer.write_all(contents).await {
                log::error!("{e}");
            }
            if let Err(e) = writer.flush().await {
                log::error!("{e}");
            }
        }
        Err(e) => log::error!("{e}"),
    }
}

// 比较器
pub struct OracleComparer {
    // 全局配置
    pub global_config: GlobalConfig,
    pub db_config: DatabaseConfig,
}

// 传入用户配置，查询表信息
// 查询表数据
// 排除大字段，如果有需要
// 计算crc并写入文件

pub enum Typ {
    SRC,
    TGT,
}

impl OracleComparer {
    pub fn new(global_config: GlobalConfig, db_config: DatabaseConfig) -> Self {
        OracleComparer {
            global_config,
            db_config,
        }
    }

    pub async fn compare(&self, ty: Typ) {
        //
        let manager = OracleManager::new(
            &self.db_config.username.clone(),
            &self.db_config.password.clone(),
            format!(
                "//{}:{}/{}",
                &self.db_config.host, &self.db_config.port, &self.db_config.database
            ),
        );

        let rs = self.global_config.clone().rule.unwrap();
        let rules;
        rules = match ty {
            Typ::SRC => rs.iter().map(|r| r.src.to_owned()).collect::<Vec<Rule>>(),
            Typ::TGT => rs.iter().map(|r| r.tgt.to_owned()).collect::<Vec<Rule>>(),
        };

        // let schemas = rules
        //     .iter()
        //     .map(|r| r.to_owned())
        //     .collect::<Vec<RuleValue>>();
        // let mut schemas = Vec::new();
        // for rule in rules {
        //     for rule_value in rule {
        //         schemas.push(rule_value.to_owned());
        //     }
        // }

        for rule in rules {
            // 数据库所有的schema
            let db_schemas = manager.checked_schemas(&rule.schema);
            // 按schema过滤
            for s in db_schemas {
                // 查询表的
                let tables = if rule.table.is_empty() {
                    // 查询所有表
                    manager.schema_tables(&s)
                } else {
                    // 查询指定表
                    manager.checked_tables(&s, &rule.table)
                };

                // let s = rule.value;
                // if !db_schemas.contains(&s) {
                //     log::error!("Non-existent schema: {}", &s);
                // }

                // 查询schema下的所有指定表
                //   1、指定table为None
                //   2、无排除表
                //   3、无包含表
                // let tables =
                //     if rule.excludes.is_none() && rule.includes.is_none() && rule.tables.is_none() {
                //         manager.schema_tables(&s)
                //     } else {
                //         // 查询匹配表
                //         //   1、含指定表
                //         //   2、含排除表
                //         //   3、含包含表
                //         let mut in_tables = Vec::new();
                //         let mut includes = match rule.includes {
                //             Some(i) => i,
                //             None => Vec::new(),
                //         };
                //         in_tables.append(&mut includes);
                //         if let Some(mut table_name) = rule.tables {
                //             in_tables.append(&mut table_name);
                //         }

                //         manager.checked_tables(
                //             &s,
                //             in_tables,
                //             match rule.excludes {
                //                 Some(e) => e,
                //                 None => Vec::new(),
                //             },
                //         )
                //     };

                // 表结构、主键、非空唯一索引时，不需要统计行数
                // let mut table_count: AHashMap<String, usize> = AHashMap::new();
                let rt = runtime::Builder::new_multi_thread()
                    .worker_threads(8)
                    .thread_name("cwkr")
                    .enable_io()
                    .enable_time()
                    .build()
                    .unwrap();

                let mut table_infos = Vec::new();
                let mut handles = vec![];
                // 所有表
                for table_name in tables {
                    let table = QueryTable::new(s.clone(), table_name.clone());
                    let cloned_manager = manager.clone();
                    match self.global_config.action {
                        Action::Primarykey
                        | Action::Columns
                        | Action::NonNullableUniqueIndex
                        | Action::Keys => {
                            let cloned_action = self.global_config.action.clone();
                            let cloned_table = table.clone();
                            let cloned_rule_column = rule.column.clone();
                            let handle = rt.spawn(async move {
                                cloned_manager.query_table_metadata(
                                    &cloned_table,
                                    &cloned_rule_column,
                                    &cloned_action,
                                )
                            });
                            handles.push(handle);
                        }
                        Action::Count => {
                            let cloned_table = table.clone();
                            let handle = rt.spawn(async move {
                                let mut table_info =
                                    TableInfo::new(cloned_table.owner, cloned_table.table_name);
                                table_info.count = cloned_manager.count(table);
                                table_info
                            });
                            handles.push(handle);
                        }
                        Action::CrcAll | Action::CrcRow => {
                            let cloned_action = self.global_config.action.clone();
                            let cloned_table = table.clone();
                            let cloned_rule_column = rule.column.clone();
                            let handle = rt.spawn(async move {
                                let count = cloned_manager.count(cloned_table.clone());
                                if count > 0 {
                                    let mut table_info = cloned_manager.query_table_metadata(
                                        &cloned_table,
                                        &cloned_rule_column,
                                        &cloned_action,
                                    );
                                    table_info.count = count;
                                    table_info
                                } else {
                                    let table_info =
                                        TableInfo::new(cloned_table.owner, cloned_table.table_name);
                                    table_info
                                }
                            });
                            handles.push(handle);
                        }
                        _ => {}
                    }
                }

                for handle in handles {
                    let now = Local::now();
                    let table_info = handle.await.unwrap();
                    let now = Local::now() - now;

                    match self.global_config.action {
                        Action::Primarykey => log::info!(
                            "Table [{}.{}] primary key fetched: {} ({:.3} sec)",
                            table_info.schema_name,
                            table_info.table_name,
                            table_info.keys.len(),
                            (now.num_milliseconds() as f64 / 1000.0) as f64
                        ),

                        Action::Columns => log::info!(
                            "Table [{}.{}] columns fetched: {} ({:.3} sec)",
                            table_info.schema_name,
                            table_info.table_name,
                            table_info.cols.len(),
                            (now.num_milliseconds() as f64 / 1000.0) as f64
                        ),
                        Action::NonNullableUniqueIndex => log::info!(
                            "Table [{}.{}] non-nullable unique index fetched: {} ({:.3} sec)",
                            table_info.schema_name,
                            table_info.table_name,
                            table_info.keys.len(),
                            (now.num_milliseconds() as f64 / 1000.0) as f64
                        ),
                        Action::Keys => log::info!(
                            "Table [{}.{}] keys fetched: {} ({:.3} sec)",
                            table_info.schema_name,
                            table_info.table_name,
                            table_info.keys.len(),
                            (now.num_milliseconds() as f64 / 1000.0) as f64
                        ),
                        _ => {}
                    };
                    table_infos.push(table_info);
                }

                rt.shutdown_background();

                match self.global_config.action {
                    Action::Primarykey
                    | Action::Columns
                    | Action::NonNullableUniqueIndex
                    | Action::Keys
                    | Action::Count => {
                        // 生成报告
                        // log::debug!("{:?}", table_infos);
                    }
                    _ => self.comp_with_crc(table_infos, &manager).await,
                }
            }
        }
    }

    // 查询数据并计算checksum
    async fn comp_with_crc(&self, table_infos: Vec<TableInfo>, manager: &OracleManager) {
        for table in table_infos {
            if table.count == 0 {
                continue;
            }

            // 针对有数据的表
            let now = Local::now();

            let rt = runtime::Builder::new_multi_thread()
                .worker_threads(8)
                .thread_name("fwkr")
                .enable_io()
                .enable_time()
                .build()
                .unwrap();

            let mut handles = vec![];
            let page_count = table.count / self.global_config.page_size
                + if table.count % self.global_config.page_size == 0 {
                    0
                } else {
                    1
                };

            /*LOCK*/
            {
                let &(ref lock, ref cvar) = &*SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                let mut map = lock.lock();
                // Acquired Lock: ReadWrite
                map.clear();
                // 通知其他等待的线程
                cvar.notify_all();
                // Released Lock
            }

            let page = Arc::new(Mutex::new(page_count));

            // let mut map  = AHashMap::new();
            loop {
                let mut lock = page.lock().await;
                let current_page: usize = lock.to_string().parse().unwrap();
                *lock -= 1;

                let cloned_table_info = table.clone();
                let cloned_manager = manager.clone();
                let global_config = self.global_config.clone();
                // row_counter,
                let handle = rt.spawn(async move {
                    query_and_checksum(
                        current_page,
                        cloned_table_info,
                        global_config,
                        &cloned_manager,
                    )
                    .await
                });
                handles.push(handle);

                if current_page <= 1 {
                    break;
                }
            }

            let mut fetched_count = 0;
            for handle in handles {
                fetched_count += handle.await.unwrap();
            }
            rt.shutdown_background();

            /*LOCK*/
            {
                let &(ref lock, ref cvar) = &*SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                let map = lock.lock();
                // Acquired Lock: ReadOnly
                // write_crc_file(self.global_config, table_info, map).await;
                let mut csv_file_names = Vec::new();
                for (key, value) in map.iter() {
                    let file_name =
                        format!("{}.{}_{}.csv", table.schema_name, table.table_name, key);

                    let csv_file_name = self.global_config.tmpdir.join(&file_name);
                    csv_file_names.push(file_name);

                    // table_info
                    //     .crc_file_names
                    //     .insert(csv_file_name.clone().to_str().unwrap().to_string(), 1);
                    let mut contents = Vec::new();
                    for (k, v) in value {
                        contents.push(format!("{},{:08x}", k, v));
                    }

                    write_contents(
                        csv_file_name,
                        format!("{}\n", contents.join("\n")).as_bytes(),
                    )
                    .await;
                }

                // 写Index文件
                let index_file_name = self
                    .global_config
                    .tmpdir
                    .join(format!("{}.{}.index", table.schema_name, table.table_name));

                write_contents(
                    index_file_name.to_path_buf(),
                    format!("{}\n", csv_file_names.join("\n")).as_bytes(),
                )
                .await;

                // 通知其他等待的线程
                cvar.notify_all();
                // Released Lock
            }

            // let file_names = crc_file_names
            //     .keys()
            //     .map(|k| k.to_owned())
            //     .collect::<Vec<String>>();

            // let index_file = self.global_config.tmpdir.join(format!(
            //     "{}.{}.index",
            //     table_info.owner, table_info.table_name
            // ));
            // if let Err(e) = fs::write(index_file.clone(), file_names.join("\n")).await {
            //     log::error!("Index file {:?} write failed, {}", index_file, e);
            // }

            let now = Local::now() - now;
            log::info!(
                "Table [{}.{}] all rows fetched {} ({:.3} sec)",
                table.schema_name,
                table.table_name,
                fetched_count,
                (now.num_milliseconds() as f64 / 1000.0) as f64
            );
        }
    }
}
